package com.avicit.zipkinkafkaesserver.dao.es;

import com.avicit.zipkinkafkaesserver.dao.es.entity.ZipkinInterfaceAverageDurationES;
import com.avicit.zipkinkafkaesserver.dao.es.entity.ZipkinInterfaceCountES;
import com.avicit.zipkinkafkaesserver.dto.ZipkinInterfaceAverageDurationDto;
import com.avicit.zipkinkafkaesserver.entity.ZipkinInterfaceAverageDuration;
import com.avicit.zipkinkafkaesserver.entity.ZipkinInterfaceDuration;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.metrics.avg.InternalAvg;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.query.UpdateQuery;
import org.springframework.data.elasticsearch.core.query.UpdateQueryBuilder;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;


@Slf4j
@Repository
public class ZipkinInterfaceAverageDurationEsDao extends BaseEsDao<ZipkinInterfaceAverageDurationES>{


    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @Override
    ElasticsearchTemplate elasticsearchTemplate() {
        return elasticsearchTemplate;
    }

    @Override
    public boolean save(String index, String type, List<ZipkinInterfaceAverageDurationES> dataList) throws IOException {
        XContentBuilder builder = getxContentBuilderMapping();
        return super.save(index, type, dataList, builder);
    }

    public List<ZipkinInterfaceAverageDurationES> query(long startTime, long endTime, long min, long max, int topN, String index, String type) {
        RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("createTime");
        rangeQueryBuilder.from(startTime).to(endTime==0?System.currentTimeMillis():endTime)
                .includeLower(true)//允许取边界值
                .includeUpper(true);

        SearchRequestBuilder searchRequestBuilder = elasticsearchTemplate.getClient().prepareSearch(index)
                .setTypes(type)
                .setQuery(rangeQueryBuilder);//时间过滤

        RangeQueryBuilder interfaceAverageDuration = QueryBuilders.rangeQuery("interfaceAverageDuration");
        //设置最大值，最小值
        if (max ==0 && min >0) {
            interfaceAverageDuration.from(min).includeLower(true);
        }
        if (max>0 && min == 0) {
            interfaceAverageDuration.to(max).includeUpper(true);
        }
        if (max>0 && min > 0) {
            interfaceAverageDuration.from(min).to(max).includeUpper(true).includeUpper(true);
        }

        searchRequestBuilder = searchRequestBuilder.setQuery(interfaceAverageDuration);

        //耗时倒排
        if (topN>0) {
            FieldSortBuilder sortBuilder = SortBuilders.fieldSort("interfaceAverageDuration").order(SortOrder.DESC);
            searchRequestBuilder = searchRequestBuilder.addSort(sortBuilder).setSize(topN);
        }
        SearchResponse response = searchRequestBuilder.execute().actionGet();

        List<ZipkinInterfaceAverageDurationES> zipkinInterfaceAverageDurations = new ArrayList<>();
        response.getHits().forEach(item->{
            Map<String, Object> sourceAsMap = item.getSourceAsMap();
            zipkinInterfaceAverageDurations.add(ZipkinInterfaceAverageDurationES.builder()
//                    .id(item.getId())
                    .interfaceMethod((String) sourceAsMap.get("interfaceMethod"))
                    .queryKey((String) sourceAsMap.get("queryKey"))
                    .interfaceIp((String) sourceAsMap.get("interfaceIp"))
                    .interfacePath((String)sourceAsMap.get("interfacePath"))
                    .interfaceMethod((String) sourceAsMap.get("interfaceMethod"))
                    .interfaceAverageDuration((Long) sourceAsMap.get("interfaceAverageDuration"))
                    .serverName((String) sourceAsMap.get("serverName"))
                    .build());
        });

        return zipkinInterfaceAverageDurations;
    }

    public ZipkinInterfaceAverageDurationES findByEndTime(String index, String type) throws IOException {
        XContentBuilder mapping = this.getxContentBuilderMapping();
        boolean create = ElasticsearchTemplateWarpper.builder(elasticsearchTemplate).createIndex(index, type, mapping);
        SearchRequestBuilder searchRequestBuilder = elasticsearchTemplate.getClient().prepareSearch(index);
        //根据 esEndTime 的最大值获取数据,根据 esEndTime 倒排，取1条
        FieldSortBuilder esEndTime = SortBuilders.fieldSort("esEndTime").order(SortOrder.DESC);
        SearchResponse response = searchRequestBuilder
                .setTypes(type)
                .addSort(esEndTime)
                .setSize(1)
                .execute().actionGet();
        ZipkinInterfaceAverageDurationES build = ZipkinInterfaceAverageDurationES.builder().build();
        response.getHits().forEach(item->{
            Map<String, Object> sourceAsMap = item.getSourceAsMap();
            build.setId(item.getId());
            build.setEsEndTime((Long) sourceAsMap.get("esEndTime"));
            build.setEsStartTime((Long) sourceAsMap.get("esStartTime"));
        });
        return build;
    }

    /**
     * 构建 mapping
     * 来自官方代码案例
     * @return
     * @throws IOException
     */
    private XContentBuilder getxContentBuilderMapping() throws IOException {
        //设置mapping,防止在新创建的index 时，还没添加数据导致的查询异常
        XContentBuilder builder = XContentFactory.jsonBuilder();
        builder.startObject();
        {
            builder.startObject("properties");
            {
                builder.startObject("batchNumber");
                {
                    builder.field("type", "long");
                }
                builder.endObject();
                builder.startObject("esEndTime");
                {
                    builder.field("type", "long");
                }
                builder.endObject();
                builder.startObject("esStartTime");
                {
                    builder.field("type", "long");
                }
                builder.endObject();
            }
            builder.endObject();
        }
        builder.endObject();
        return builder;
    }


    /**
     * 求 接口的平均耗时
     * @param index
     * @param type
     * @return
     */
    public List<ZipkinInterfaceAverageDurationDto> getCountResultFromEs(String index, String type) {
        ElasticsearchTemplateWarpper.builder(elasticsearchTemplate).createIndex(index);
        SearchResponse response = elasticsearchTemplate.getClient()
                .prepareSearch(index)
                .setTypes(type)
                .addAggregation(AggregationBuilders.terms("interfaceKey").field("interfaceKey.keyword")//分组
                        .subAggregation(AggregationBuilders.avg("avg").field("interfaceAverageDuration"))//平均
                )
                .execute().actionGet();
        List<Aggregation> aggregations = response.getAggregations().asList();
        List<ZipkinInterfaceAverageDurationDto> zipkinInterfaceAverageDurationDtos = new ArrayList<>();
        aggregations.forEach(item->{
            if (item instanceof StringTerms) {
            StringTerms stringTerms = (StringTerms) item;
            List<StringTerms.Bucket> buckets = stringTerms.getBuckets();

            buckets.forEach(bucket -> {
                ZipkinInterfaceAverageDurationDto averageDurationDto = ZipkinInterfaceAverageDurationDto.builder().build();
                String key = (String)bucket.getKey();
                InternalAvg avg = bucket.getAggregations().get("avg");
                double value = avg.getValue();
                averageDurationDto.setInterfaceKey(key);
                averageDurationDto.setInterfaceAverageDuration((long)value);
                zipkinInterfaceAverageDurationDtos.add(averageDurationDto);
            });
        }});
        log.info("接口的平均耗时 统计 :{}", zipkinInterfaceAverageDurationDtos);
        //倒排
        return zipkinInterfaceAverageDurationDtos.stream().sorted(Comparator.comparing(ZipkinInterfaceAverageDurationDto::getInterfaceAverageDuration).reversed()).collect(Collectors.toList());
    }

    /**
     * 多条件查询
     * @param startTime
     * @param endTime
     * @param serverName
     * @param interfacePath
     * @param index
     * @return
     */
    public List<ZipkinInterfaceAverageDurationES> getRealtimeAvgDurationChartData(long startTime, long endTime, String serverName, String interfacePath, String index, String type) {

        SearchRequestBuilder searchRequestBuilder = elasticsearchTemplate.getClient().prepareSearch(index);

        FieldSortBuilder queryStartTime = SortBuilders.fieldSort("queryStartTime").order(SortOrder.DESC);

        RangeQueryBuilder queryStartTimeRang = QueryBuilders.rangeQuery("queryStartTime");
        RangeQueryBuilder queryEndTimeRand = QueryBuilders.rangeQuery("queryEndTime");
        queryStartTimeRang.from(startTime)
                .includeLower(true);//允许取边界值
        queryEndTimeRand.to(endTime).includeUpper(true);


        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchPhraseQuery("serverName", serverName))
                .must(QueryBuilders.matchPhraseQuery("interfacePath", interfacePath))
                .must(queryStartTimeRang)
                .must(queryEndTimeRand);

        SearchResponse response = searchRequestBuilder
                .setQuery(boolQueryBuilder)
                .setTypes(type)
                .addSort(queryStartTime)
                .execute()
                .actionGet();
        SearchHits hits = response.getHits();
        List<ZipkinInterfaceAverageDurationES> result = new ArrayList<>();
        hits.forEach(item->{
            Map<String, Object> sourceAsMap = item.getSourceAsMap();
            int interfaceAverageDuration =(int) sourceAsMap.get("interfaceAverageDuration");
            ZipkinInterfaceAverageDurationES build = ZipkinInterfaceAverageDurationES.builder()
                    .esStartTime((Long) sourceAsMap.get("esStartTime"))
                    .queryStartTime((Long) sourceAsMap.get("queryStartTime"))
                    .queryEndTime((Long) sourceAsMap.get("queryEndTime"))
                    .interfaceIp((String) sourceAsMap.get("interfaceIp"))
                    .interfacePath((String) sourceAsMap.get("interfacePath"))
                    .serverName((String) sourceAsMap.get("serverName"))
                    .interfaceAverageDuration(Long.valueOf(interfaceAverageDuration)).build();
            result.add(build);
        });

        return result;
    }
}
